package com.ledou44.rocketmqlearn.util;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.Collectors;

@Component
public class RocketMqUtil {

    private static DefaultMQProducer client;

    private static DefaultMQPushConsumer consumer;

    private static final String TOPIC = "defaultTopic";

    @Autowired
    public void setClient(DefaultMQProducer client) {
        RocketMqUtil.client = client;
    }

    @Autowired
    public void setConsumer(DefaultMQPushConsumer consumer) {
        RocketMqUtil.consumer = consumer;
    }

    public static void pushMessage(String str) {
        try {
            client.start();
            Message message = new Message(TOPIC, "tag", "keys",str.getBytes(StandardCharsets.UTF_8));
            client.send(message);
            client.shutdown();
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void startConsumer() {
        try {
            consumer.subscribe(TOPIC, "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext contest) {
                    List<String> result = list.stream().map(item -> new String(item.getBody(), StandardCharsets.UTF_8)).collect(Collectors.toList());
                    System.out.println(JSON.toJSONString(result));
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

}
